package org.raftkv.raft.tracker;

/**
 * Inflights 限制了发送给 Follower 但尚未被 ack 的 MsgApp（由其中包含的最大索引表示）的数量。
 * <br/>
 * 调用者使用 full() 检查是否可以发送更多消息，在发送新附加项时调用 add()，并在收到确认时通过 freeLE() 函数释放配额
 *
 * @author wzy
 */
public class Inflights {

    /**
     * 缓冲区中的起始索引
     */
    private int start;

    /**
     * 缓冲区的 Inflight 数量
     */
    private int count;

    /**
     * 缓冲区的大小
     */
    private int size;

    /**
     * 缓冲区包含最后一个消息中的条目索引
     */
    private long[] buffer;

    private Inflights(int size) {
        this.size = size;
        this.buffer = new long[size];
    }

    /**
     * 新建一个 Inflights，允许最多 size 个 Inflight 消息
     */
    public static Inflights newInflights(int size) {
        return new Inflights(size);
    }

    @Override
    public Inflights clone() {
        Inflights ins = new Inflights(this.size);
        ins.start = this.start;
        ins.count = this.count;
        System.arraycopy(this.buffer, 0, ins.buffer, 0, this.buffer.length);
        return ins;
    }

    /**
     * 通知 Inflights 正在发送具有给定索引的新消息。
     * <br/>
     * 在 add() 之前必须调用 full() 来验证是否还有空间容纳一个以上消息，并且连续调用 add() 必须提供单调的索引序列。
     */
    public void add(long inflight) {
        if (this.full()) {
            throw new RuntimeException("cannot add into a Full inflights");
        }
        int next = this.start + this.count;
        int size = this.size;
        if (next >= size) {
            next -= size;
        }
        if (next >= this.buffer.length) {
            this.grow();
        }
        this.buffer[next] = inflight;
        this.count++;
    }

    /**
     * grow 函数翻倍增加 Inflight 缓冲区大小，直到达到 Inflight.size。
     * <br/>
     * 我们根据需要 grow()，而不是直接达到预先分配的 Inflights.size，以便支撑每个进程有数千个 Raft 组的系统
     */
    private void grow() {
        int newSize = this.buffer.length * 2;
        if (newSize == 0) {
            newSize = 1;
        } else if (newSize > this.size) {
            newSize = this.size;
        }
        long[] newBuffer = new long[newSize];
        System.arraycopy(this.buffer, 0, newBuffer, 0, this.buffer.length);
        this.buffer = newBuffer;
    }

    /**
     * 释放小于等于 to 的 Inflight
     */
    public void freeLE(long to) {
        if (this.count == 0 || to < this.buffer[this.start]) {
            // 超出了窗口左侧
            return;
        }

        int idx = this.start;
        int i;
        for (i = 0; i < this.count; i++) {
            // 找到了第一个大的 Inflight
            if (to < this.buffer[idx]) {
                break;
            }

            // 增加索引并且可能旋转
            int size = this.size;
            if (++idx >= size) {
                idx -= size;
            }
        }
        // 释放 i 个 Inflight 并设置新的 index
        this.count -= i;
        this.start = idx;
        if (this.count == 0) {
            // InFlight 为空，重置起始索引，以免不必要地增加缓冲区
            this.start = 0;
        }
    }

    /**
     * 释放第一个 Inflight，如果 Inflight 中没东西，则这个是空操作
     */
    public void freeFirstOne() {
        this.freeLE(this.buffer[this.start]);
    }

    /**
     * 如果目前无法发送更多消息，则干函数返回 true
     */
    public boolean full() {
        return this.count == this.size;
    }

    /**
     * 返回 Inflight 消息的数量
     */
    public int count() {
        return this.count;
    }

    /**
     * 该函数释放所有的 Inflights
     */
    public void reset() {
        this.count = 0;
        this.start = 0;
    }
}
